package com.disthen.message.core.converter;

import org.apache.thrift.TBase;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.SimpleMessageConverter;

import com.disthen.message.core.thrift.model.message;

public class ThriftSupportConverter extends SimpleMessageConverter {
	
	private Class<?> messageBodyType = message.class;
	
	public static final String CONTENT_TYPE_THRIFT_OBJECT = "application/x-thrift-object";

	public void setMessageBodyType(Class<?> messageBodyType) {
		this.messageBodyType = messageBodyType;
	}

	@Override
	protected Message createMessage(Object object,
			MessageProperties messageProperties) {
		byte[] bytes = null;
		if (object instanceof byte[]) {
			bytes = (byte[]) object;
			messageProperties.setContentType(MessageProperties.CONTENT_TYPE_BYTES);
		}
		else if (object instanceof TBase) {
			TSerializer serializer = new TSerializer();
			try {
				bytes = serializer.serialize((TBase<?,?>) object);
			} catch (TException e) {
				throw new MessageConversionException(
						"failed to convert to serialized Message content", e);
			}
			messageProperties.setContentType(CONTENT_TYPE_THRIFT_OBJECT);
		}
		if (bytes != null) {
			messageProperties.setContentLength(bytes.length);
			return new Message(bytes, messageProperties);
		}
		else {
			return super.createMessage(object, messageProperties);
		}
	}

	@Override
	public Object fromMessage(Message message)
			throws MessageConversionException {
		MessageProperties props = message.getMessageProperties();
		if(props.getContentType().equals(CONTENT_TYPE_THRIFT_OBJECT)){
			TDeserializer deserializer = new TDeserializer();
			Object msg = null;
			try {
				msg = messageBodyType.newInstance();
				deserializer.deserialize((TBase<?,?>) msg, message.getBody());
				return msg;
			} catch (InstantiationException e1) {
				return null;
			} catch (IllegalAccessException e1) {
				return null;
			}
			catch (TException e) {
				return super.fromMessage(message);
			}
		}
		else return super.fromMessage(message);
	}

}
